package health

import (
	"context"
	"gitee.com/zaper/kafka-service/service/connect"
	metamgr "gitee.com/zaper/kafka-service/service/metadata"
	logger "gitee.com/zaper/private-logger"
	"github.com/segmentio/kafka-go"
	"net"
	"time"
)

func CheckReady() {
	logger.Info("begin check kafka net con")
	retry(checkNet, 2)
	logger.Info("kafka net con success")

	logger.Info("begin check kafka controller")
	retry(checkController, 2)
	logger.Info("kafka controller ready")

	logger.Info("begin check kafka broker")
	retry(checkBroker, 2)
	logger.Info("kafka broker ready")

	logger.Info("begin check kafka system data")
	retry(checkSystemData, 2)
	logger.Info("kafka system data ok")
}

func checkNet() bool {
	for _, addr := range connect.Brokers {
		if !func() bool {
			conn, err := net.DialTimeout("tcp", addr, 2*time.Second)
			defer func() {
				if conn != nil {
					e := conn.Close()
					if e != nil {
						return
					}
				}
			}()
			if err != nil {
				return false
			}
			return true
		}() {
			return false
		}
	}
	return true
}

func checkController() bool {
	controller := metamgr.GetMetadataManager().GetController()
	if controller != nil && controller.ID != -1 {
		return true
	}
	return false
}

func checkBroker() bool {
	broker := metamgr.GetMetadataManager().GetAllBrokers()
	if broker != nil && len(broker) != 0 {
		return true
	}
	return false
}

func checkSystemData() bool {
	_, err := connect.GetKafkaConnectManager().Client.Metadata(context.Background(), &kafka.MetadataRequest{
		Topics: []string{"__consumer_offsets"},
	})
	if err != nil {
		return false
	}
	return true
}

func retry(function func() bool, dy int) bool {
	for {
		if function() {
			return true
		}
		time.Sleep(time.Duration(dy) * time.Second)
	}
}
